package com.kin207.netty.message.mqtt.listener;



import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.kin207.netty.handle.IMessageHandle;
import com.kin207.netty.listener.ListenerConfig;
import com.kin207.netty.listener.MessageHandleListener;
import com.kin207.netty.util.StringParseUtils;

import com.kin207.netty.message.mqtt.MqttMessageCallbackListener;
import com.kin207.netty.message.mqtt.service.LockMqttService;
public class MqttMessageListener  implements  MessageHandleListener{
	final static Logger logger = LoggerFactory.getLogger(MqttMessageListener.class);
	private String clientId ="admin01";
	private String serverURI="tcp://127.0.0.1:1883";
	private String messageType="json";
	private String Qos;
	private int[] Qoss = {1,1,1,1};
	private String relayQuauaName;
	private String[] relayQuauaNames=new String[]{"Q.MSG.SEND.NETTY.START,Q.MSG.SEND.NETTY.ANSWER,Q.MSG.SEND.NETTY.POST.OPENLOCK,Q.MSG.SEND.NETTY.POST.GENERAL"};
	private MqttClient mqttClient;

	public static  MqttMessageListener messageListener;
	public void initialize(ListenerConfig cfg) {
		StringParseUtils spu = new StringParseUtils();
		
		if (!spu.set(cfg.getParam("relayQuauaNames")).isEmpty()) {
			relayQuauaName = spu.get();
			logger.info("conf relayQuauaNames:{}.", relayQuauaName);
			String[] relayQuauas = relayQuauaName.split(",");
			relayQuauaNames = relayQuauas;
		}
		if (!spu.set(cfg.getParam("Qos").toUpperCase()).isEmpty()) {
			Qos = spu.get();
			logger.info("set Qos:{}.", Qos);
		}
		String[] Qossplit = Qos.split(",");
		for (int i = 0; i < Qossplit.length; i++) {
			Qoss[i] = Integer.valueOf(Qossplit[i].trim());
		}
		if (!spu.set(cfg.getParam("clientId")).isEmpty()) {
			clientId = spu.get();
			logger.info("set clientId:{}.", clientId);
		}

		if (!spu.set(cfg.getParam("serverURI")).isEmpty()) {
			serverURI = spu.get();
			logger.info("set serverURI:{}.", serverURI);
		}

		if (!spu.set(cfg.getParam("messageType")).isEmpty()) {
			messageType = spu.get();
			logger.info("set messageType:{}.", messageType);
		}

		MqttConnectOptions mqcConf = new MqttConnectOptions();
		if (!spu.set(cfg.getParam("options")).isEmpty()) {
			String options = spu.get();
			mqcConf = JSONObject.parseObject(options, MqttConnectOptions.class);
		}
		this.connect(mqcConf);
	}
	/**
	 * @param mqcConf
	 */
	public void connect(MqttConnectOptions mqcConf) {
		try {
			if (mqttClient == null) {
				synchronized (MqttClient.class) {
					if (mqttClient == null) {
						mqttClient = new MqttClient(serverURI, clientId, new MemoryPersistence());
					}
				}
			}
			logger.error("open mqtt connection, options:{}", mqcConf, new MemoryPersistence());
			if (!mqttClient.isConnected()) {
				mqttClient.setCallback(new MqttMessageCallbackListener(new LockMqttService()));
				mqttClient.connect(mqcConf);
				subscribe(relayQuauaNames);
				
			} else {
				mqttClient.disconnect();
				mqttClient.setCallback(new MqttMessageCallbackListener(new LockMqttService()));
				mqttClient.connect(mqcConf);
				subscribe(relayQuauaNames);
				logger.error("open mqtt reconnection success.");
			}
			logger.info("open mqtt connection success.");
		} catch (MqttException e) {
			logger.error(e.getMessage(), e);
		}
	}
	private void subscribe(String[] relayQuauaNames) {
		try {
			logger.debug("subscribe(String[] relayQuauaNames relayQuauaNames.length)");
			mqttClient.subscribe(relayQuauaNames, Qoss);
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
			throw new RuntimeException(e.getMessage(), e);
		}
	}
	
	public String getClientId() {
		return clientId;
	}

	public void setClientId(String clientId) {
		this.clientId = clientId;
	}


	public String getServerURI() {
		return serverURI;
	}

	public void setServerURI(String serverURI) {
		this.serverURI = serverURI;
	}

	public String getMessageType() {
		return messageType;
	}

	public void setMessageType(String messageType) {
		this.messageType = messageType;
	}



	public String getQos() {
		return Qos;
	}
	public void setQos(String qos) {
		Qos = qos;
	}
	public int[] getQoss() {
		return Qoss;
	}
	public void setQoss(int[] qoss) {
		Qoss = qoss;
	}
	public MqttClient getMqttClient() {
		return mqttClient;
	}

	public void setMqttClient(MqttClient mqttClient) {
		this.mqttClient = mqttClient;
	}
	public void created(IMessageHandle t) {
		if (relayQuauaNames == null && relayQuauaNames.length==0) {
			logger.debug("message filter, not send.");
			return;
		}
	}
	public void destroyed(IMessageHandle t) {
		// TODO Auto-generated method stub
		
	}
}
